package net.xdclass.base_project.consumer;

import net.xdclass.base_project.config.SpringConfigConstant;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;

import java.util.List;

/**
 * @author sqz
 * @Description: kafka多分区监听器
 * @date 2019/9/26 11:06
 */
public class KafkaPartitionLintener {

    private final Logger log = LoggerFactory.getLogger(KafkaPartitionLintener.class);

    @KafkaListener(groupId = SpringConfigConstant.KAFKA_GROUP_ID, topicPartitions = { @TopicPartition(topic = SpringConfigConstant.KAFKA_FIRST_TOPIC, partitions = { "0" }) })
    public void listenPartition0(List<ConsumerRecord<String, String>> records) {
        records.forEach(record->{
            log.info("id0--message{}:" +  String.valueOf(record.value()));
        });
    }

    @KafkaListener(groupId = SpringConfigConstant.KAFKA_GROUP_ID, topicPartitions = { @TopicPartition(topic = SpringConfigConstant.KAFKA_FIRST_TOPIC, partitions = { "1" }) })
    public void listenPartition1(List<ConsumerRecord<String, String>> records) {
        records.forEach(record->{
            log.info("id1--message{}:" +  String.valueOf(record.value()));

        });
    }
}

